package com.gaofans.nrpc.server.bootstrap;

import com.gaofans.nrpc.common.context.SelectableExecutorGroup;
import com.gaofans.nrpc.common.context.SelectableExecutor;
import com.gaofans.nrpc.common.exception.ExceptionHandler;
import com.gaofans.nrpc.common.protocol.EchoMessageHandler;
import com.gaofans.nrpc.common.protocol.MessageCodec;
import com.gaofans.nrpc.common.protocol.ProtocolFrameDecoder;
import com.gaofans.nrpc.server.handler.ContextBindHandler;
import com.gaofans.nrpc.server.handler.RequestInFilterHandler;
import com.gaofans.nrpc.server.handler.RpcRequestHandler;
import com.gaofans.nrpc.server.handler.ServerFilter;
import com.gaofans.nrpc.server.service.ServiceBinder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

import javax.net.ssl.SSLEngine;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author GaoFans
 */
@SuppressWarnings({"rawtypes","unchecked"})
public class NettyRpcServer implements RpcServer{

    private static final Logger LOGGER = LoggerFactory.getLogger(NettyRpcServer.class);

    private final ServerBootstrapConfig config;

    private Channel channel;

    private Boolean isStart = false;

    private final ServiceBinder binder;

    private final List<ServerFilter> filters;

    private NioEventLoopGroup parentGroup;

    private NioEventLoopGroup childGroup;

    private DefaultEventLoopGroup worker;

    private SelectableExecutor chooser;

    public NettyRpcServer(ServerBootstrapConfig config, ServiceBinder binder) {
        this.config = config;
        this.binder = binder;
        this.filters = new ArrayList<>();
    }

    @Override
    public Mono<Void> start() {
        if (isStart){
            throw new IllegalStateException("服务器已经启动");
        }
        return Mono.create(sink -> {
            parentGroup = new NioEventLoopGroup(config.getParentThreadCount());
            childGroup = new NioEventLoopGroup(config.getChildThreadCount());
            worker = new DefaultEventLoopGroup(config.getWorkerThreadCount());
            chooser = new SelectableExecutorGroup(config.getHandlerThreadCount());
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(parentGroup, childGroup)
                    .channel(NioServerSocketChannel.class);
            Map<ChannelOption, Object> optionMap = config.getOptionMap();
            if (optionMap != null && optionMap.size() > 0){
                for (Map.Entry<ChannelOption, Object> entry : optionMap.entrySet()) {
                    bootstrap.option(entry.getKey(),entry.getValue());
                }
            }
            LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);
            MessageCodec messageCodec = new MessageCodec(config.getSerializerFactory(), config.getSerializerType(), config.getVersion(), config.getCodec());
            ContextBindHandler contextBindHandler = new ContextBindHandler(chooser);
            RpcRequestHandler rpcRequestHandler = new RpcRequestHandler(binder);
            EchoMessageHandler echoMessageHandler = new EchoMessageHandler(config.getIdGenerator());
            bootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {

                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    if (!Objects.isNull(config.getSslContext())){
                        SSLEngine engine = config.getSslContext().newEngine(ch.alloc());
                        engine.setUseClientMode(false);
                        engine.setNeedClientAuth(true);
                        ch.pipeline().addLast(new SslHandler(engine));
                    }
                    ch.pipeline().addLast(new ProtocolFrameDecoder());
                    ch.pipeline().addLast(loggingHandler);
                    ch.pipeline().addLast(messageCodec);
                    ch.pipeline().addLast(new IdleStateHandler(config.getReaderIdleTimeSecond(), config.getWriterIdleTimeSecond(), 0));
                    ch.pipeline().addLast(echoMessageHandler);
                    ch.pipeline().addLast(contextBindHandler);
                    ch.pipeline().addLast(worker,new RequestInFilterHandler(filters));
                    ch.pipeline().addLast(worker,rpcRequestHandler);
                    ch.pipeline().addLast(new ExceptionHandler());

                }
            });
            bootstrap.bind(config.getPort()).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()){
                    isStart = true;
                    channel = future.channel();
                    sink.success();
                }else {
                    LOGGER.error("server start failed",future.cause());
                    sink.error(future.cause());
                }
            });
        }).doOnError(throwable -> {
            executorsShutdown();
        }).doOnSuccess(unused -> {
            LOGGER.info("server start in port:{}.",config.getPort());
        }).cast(Void.class);
    }

    @Override
    public Mono<Void> stop() {
        if (!isStart || channel == null){
            throw new IllegalStateException("服务器尚未启动");
        }
        return Mono.create(sink -> {
            executorsShutdown();
            channel.close().addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()){
                    isStart = false;
                    sink.success();
                }else {
                    LOGGER.error("server stop failed",future.cause());
                    sink.error(future.cause());
                }
            });
        });
    }

    @Override
    public <T> void registerService(Class<T> serviceClass, T service) {
        binder.register(serviceClass,service);
    }

    @Override
    public void addFilter(ServerFilter filter) {
        this.filters.add(filter);
    }

    private void executorsShutdown(){
        if (worker != null){
            worker.shutdownGracefully().addListener((GenericFutureListener<Future<Object>>) future -> {
                if (future.isSuccess()){
                    LOGGER.debug("worker group closed");
                }else {
                    LOGGER.debug("worker group close failed");
                }
            });
        }
        if (childGroup != null){
            childGroup.shutdownGracefully().addListener((GenericFutureListener<Future<Object>>) future -> {
                if (future.isSuccess()){
                    LOGGER.debug("child group closed");
                }else {
                    LOGGER.debug("child group close failed");
                }
            });
        }
        if (parentGroup != null){
            parentGroup.shutdownGracefully().addListener((GenericFutureListener<Future<Object>>) future -> {
                if (future.isSuccess()){
                    LOGGER.debug("parent group closed");
                }else {
                    LOGGER.debug("parent group close failed");
                }
            });
        }
        if (chooser != null){
            chooser.shutdownGracefully().addListener((GenericFutureListener<Future<Object>>) future -> {
                if (future.isSuccess()){
                    LOGGER.debug("chooser group closed");
                }else {
                    LOGGER.debug("chooser group close failed");
                }
            });
        }
    }
}
